package com.doit.day03;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.Properties;

public class _01_消费者组在均衡策略现象的观察4 {
        private static final String HOST = "linux01:9092,linux02:9092,linux03:9092";
        private static final String GROUP_ID = "G02";
        public static void main(String[] args) {
            Properties props = new Properties();
            props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());
            props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, _01_消费者组在均衡策略现象的观察4.HOST);
            props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, _01_消费者组在均衡策略现象的观察4.GROUP_ID);

            props.setProperty(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG,"true");
            props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
            props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
            //加一个参数  消费者组在均衡策略的选择参数
            props.setProperty(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, CooperativeStickyAssignor.class.getName());

            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

            consumer.subscribe(Arrays.asList("reb-1", "reb-2"), new ConsumerRebalanceListener() {
                /*
                指的是：回收之前该消费者消费的所有分区的数据
                 */
                @Override
                public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                    System.out.println("回收之前的分区："+partitions);
                }

                /*
                 * 再分配之后，该消费者消费的所有分区的数据
                 */
                @Override
                public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                    System.out.println("分配完成之后消费的分区："+partitions);
                }
            });

            while (true){
                ConsumerRecords<String, String> poll = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));
            }
    }
}
